package com.boot2.core.conf.mqtt;

import com.boot2.core.mqtt.HlPahoMessageConverter;
import com.boot2.core.mqtt.MqttOutboundProperties;
import com.boot2.core.mqtt.MqttProperties;
import lombok.extern.apachecommons.CommonsLog;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

/**
 * @author zhangweilin
 * @date 2018/9/20 11:38
 * @description: mqtt发消息配置
 */
@Configuration
@CommonsLog
@Conditional(MqttOutboundConditional.class)
public class MqttOutboundConfiguration extends MqttConf {

    @Autowired
    private MqttProperties mqttProperties;

    @Qualifier("outoundPahoMessageConverter")
    @Autowired(required = false)
    private HlPahoMessageConverter hlPahoMessageConverter = new HlPahoMessageConverter();

    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() throws Exception {
        log.info("outBound.mqttProperties = " + mqttProperties);
        MqttOutboundProperties outboundProperties = mqttProperties.getOutbound();
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttProperties.getClientId() + "_Outbound",
                mqttClientFactory(mqttProperties, outboundProperties.getCleanSession()));
        messageHandler.setAsync(outboundProperties.getAsync());
        messageHandler.setDefaultQos(outboundProperties.getDefaultQos());
        messageHandler.setDefaultRetained(outboundProperties.getDefaultRetained());
        messageHandler.setShouldTrack(outboundProperties.getShouldTrack());
        messageHandler.setAsyncEvents(outboundProperties.getAsyncEvents());
        messageHandler.setCountsEnabled(outboundProperties.getCountsEnabled());
        messageHandler.setLoggingEnabled(outboundProperties.getLoggingEnabled());
        messageHandler.setStatsEnabled(outboundProperties.getStatsEnabled());
        messageHandler.setDefaultTopic(outboundProperties.getDefaultTopic());
        messageHandler.setConverter(hlPahoMessageConverter);
        messageHandler.setCompletionTimeout(outboundProperties.getCompletionTimeout());
        return messageHandler;
    }

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

}
